package com.bw.gmall.realtime.dim.function;


import com.alibaba.fastjson.JSONObject;
import com.bw.gmall.realtime.common.bean.TableProcessDim;
import com.bw.gmall.realtime.common.util.JdbcUtil;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.util.HashMap;
import java.util.List;

public class dim_processfunction extends BroadcastProcessFunction<JSONObject, TableProcessDim, Tuple2<JSONObject,TableProcessDim>> {


    private final MapStateDescriptor<String, TableProcessDim> mapStateDescriptor;
    private HashMap<String, TableProcessDim> hashMap;

    //构造 状态
    public dim_processfunction(MapStateDescriptor<String, TableProcessDim> mapStateDescriptor) {
        this.mapStateDescriptor = mapStateDescriptor;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        //处理 广播流 迟到问题 导致不匹配
        hashMap = new HashMap<>();
        java.sql.Connection mysqlConnection = JdbcUtil.getMysqlConnection();

        List<TableProcessDim> tableProcessDims = JdbcUtil.queryList(mysqlConnection, "select * from gmall2023_config.table_process_dim", TableProcessDim.class, true);

        //获取 数据 到hashmap
        for (TableProcessDim tableProcessDim : tableProcessDims) {
            hashMap.put(tableProcessDim.getSourceTable(),tableProcessDim);
        }

    }

    @Override
    public void processElement(JSONObject jsonObject, BroadcastProcessFunction<JSONObject, TableProcessDim, Tuple2<JSONObject, TableProcessDim>>.ReadOnlyContext readOnlyContext, Collector<Tuple2<JSONObject, TableProcessDim>> collector) throws Exception {
        //主流
        ReadOnlyBroadcastState<String, TableProcessDim> broadcastState = readOnlyContext.getBroadcastState(mapStateDescriptor);

        //获取主流中的表名
        String table = jsonObject.getString("table");
        TableProcessDim tableProcessDim = broadcastState.get(table);

        if(tableProcessDim==null){
            //分两种情况 1.是事实表数据 2. 因为 广播流 迟到导致无法匹配
            tableProcessDim= hashMap.get(table);

        }

        if(tableProcessDim!=null){
            //关联到的数据
            System.out.println("jsonObject1 = " + jsonObject);
            System.out.println("tableProcessDim1 = " + tableProcessDim);
            collector.collect(Tuple2.of(jsonObject,tableProcessDim));
        }

    }

    @Override
    public void processBroadcastElement(TableProcessDim tableProcessDim, BroadcastProcessFunction<JSONObject, TableProcessDim, Tuple2<JSONObject, TableProcessDim>>.Context context, Collector<Tuple2<JSONObject, TableProcessDim>> collector) throws Exception {
        //广播流
        BroadcastState<String, TableProcessDim> broadcastState = context.getBroadcastState(mapStateDescriptor);

        broadcastState.put(tableProcessDim.getSourceTable(),tableProcessDim);

        //若 删除表 也要删除 相应的 状态数据
        String op = tableProcessDim.getOp();

        if("d".equals(op)){
            broadcastState.remove(tableProcessDim.getSourceTable());

            hashMap.remove(tableProcessDim.getSourceTable());
        }


    }

}
